package org.ngbx.demo.flink.kafka.connector;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 连接Kafka的Demo.
 */
public class KafkaConnectorDemoJob {
    private static final Logger log = LoggerFactory.getLogger(KafkaConnectorDemoJob.class);

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KafkaSource<KafkaMessage> source = KafkaSource.<KafkaMessage>builder()
                .setBootstrapServers("kafka:9092")
                .setTopics("test_topic")
                .setGroupId("flink-kafka-demo")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new KafkaMessageSchema())
                .build();

        DataStream<KafkaMessage> ds = env.fromSource(source, WatermarkStrategy.noWatermarks(), "My Kafka Source");
        ds.map((MapFunction<KafkaMessage, KafkaMessage>) message -> {
            log.info("received message: " + message);
            return message;
        });

        env.execute("Kafka Job");
    }
}